package spider;

import annotation.ClassXpath;
import annotation.FieldXpath;
import bean.DBParams;
import bean.DownloadParams;
import bean.FetchParams;
import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
import com.alibaba.rocketmq.client.producer.DefaultMQProducer;
import common.Constant;
import common.Util;
import lombok.Data;
import org.apache.commons.lang3.text.WordUtils;
import org.jsoup.Jsoup;
import org.jsoup.select.Elements;
import org.openqa.selenium.JavascriptExecutor;
import org.openqa.selenium.WebDriver;
import org.openqa.selenium.phantomjs.PhantomJSDriver;
import org.openqa.selenium.phantomjs.PhantomJSDriverService;
import org.openqa.selenium.remote.CapabilityType;
import org.openqa.selenium.remote.DesiredCapabilities;
import org.openqa.selenium.support.ui.WebDriverWait;
import org.yaml.snakeyaml.Yaml;
import us.codecraft.xsoup.Xsoup;

import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.lang.annotation.Annotation;
import java.lang.reflect.Field;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.concurrent.*;
import java.util.function.Function;
import java.sql.*;

import static common.Constant.PAGE_DEPTH;
import static common.Constant.PROCESSOR_NUM;

/**
 * @author chen_
 * @date 2017/5/2
 */
@Data
public class Spider {
    private int index = 0;      //tasks cursor && queues cursor

    private LinkedBlockingQueue[] queues = new LinkedBlockingQueue[PAGE_DEPTH];


    private Thread[] tasks = new Thread[PAGE_DEPTH];
    private Future[] futures = new Future[PAGE_DEPTH];


    private ExecutorService taskExecutor = Executors.newFixedThreadPool(PROCESSOR_NUM);

    private Statement stmt;

    //初始化最开始的url队列
    public void setQueues0(LinkedBlockingQueue queues0) {
        queues[index] = queues0;
    }

    public static WebDriver initWebDriver() {
        DesiredCapabilities capabilities = DesiredCapabilities.phantomjs();
        capabilities.setCapability("phantomjs.binary.path", Constant.PHANTOMJS_PATH);
        List cliArgsCap = new ArrayList<String>();
        cliArgsCap.add("--web-security=false");
        cliArgsCap.add("--ssl-protocol=any");
        cliArgsCap.add("--ignore-ssl-errors=true");
        cliArgsCap.add("--webdriver-loglevel=INFO");
        cliArgsCap.add("--load-images=false");
        capabilities.setCapability(CapabilityType.SUPPORTS_FINDING_BY_CSS, true);
        capabilities.setCapability(CapabilityType.TAKES_SCREENSHOT, true);
        capabilities.setCapability(PhantomJSDriverService.PHANTOMJS_CLI_ARGS, cliArgsCap);
        return new PhantomJSDriver(capabilities);
    }


    WebDriver driver = null;

    public boolean isLastThreadLive(int k) {
        if (k > 0) {
            return !futures[k - 1].isDone();
        }
        return false;
    }

    private Spider run(Function function) {
        //当前队列
        if (queues[index] == null) {
            queues[index] = new LinkedBlockingQueue();

        }
        //下一个队列
        if (queues[index + 1] == null) {
            queues[index + 1] = new LinkedBlockingQueue();
        }
        int k = index;
        tasks[index] = new Thread(
                () -> {
                    while (!queues[k].isEmpty() || isLastThreadLive(k)) {

                        if (!queues[k].isEmpty()) {
                            try {
                                Object obj = queues[k].take();
                                if ((boolean) function.apply(obj)) {
                                    queues[k + 1].put(obj);
                                }
                            } catch (InterruptedException e) {
                                System.out.println(e.getMessage());
                            }
                        }
                    }
                }
        );
        index++;
        return this;
    }

    int j = 0;

    public Spider print() {
        return run(
                (obj) -> {
                    System.out.println(obj + "\t" + j++);
                    return true;
                }
        );
    }

    public Spider filter(Function func) {
        return run(
                (obj) -> {
                    return (boolean) func.apply(obj);
                }
        );
    }

    public Spider download(Function func) {
        return run((obj) -> {
            DownloadParams downloadParams = null;
            downloadParams = (DownloadParams) func.apply(obj);
            Util.download(downloadParams.getUrl(), downloadParams.getFilename());
            return true;
        });
    }

    public void end() {
        for (int i = 0; i < index; i++) {
            futures[i] = taskExecutor.submit(tasks[i]);
        }
        System.out.println("启动爬虫……");
        taskExecutor.shutdown();

        try {
            taskExecutor.awaitTermination(Long.MAX_VALUE, TimeUnit.MINUTES);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        close();
    }

    private void close() {
        try {
            if (stmt != null) {
                conn.close();
            }
        } catch (SQLException se) {
        }// do nothing
        try {
            if (conn != null) {
                conn.close();
            }
        } catch (SQLException se) {
            se.printStackTrace();
        }
    }

    private static boolean flag = true;

    public Spider save(Class clazz) {
        //限制初始化一次
        synchronized (this) {
            if (flag) {
                initDB();
                flag = false;
            }
        }
        return run((obj) -> {
            String tableName = clazz.getSimpleName();
            List<String> insertNames = new ArrayList<>();
            List<String> insertValues = new ArrayList<>();
            for (Field field : clazz.getDeclaredFields()) {
                insertNames.add(field.getName().substring(field.getName().lastIndexOf('.') + 1));
                try {
                    field.setAccessible(true);
                    Object o = field.get(obj);
                    if (!field.getType().isAssignableFrom(Integer.class) && o != null) {
                        insertValues.add("'" + o + "'");
                    } else {
                        insertValues.add((String) o);
                    }
                } catch (IllegalAccessException e) {
                    e.printStackTrace();
                }
            }
            String sql = "insert into " + tableName + "(" + "".join(",", insertNames) + ")" + " values " + "(" + "".join(",", insertValues) + ");";
            try {
                System.out.println(sql);
                stmt.executeUpdate(sql);
            } catch (SQLException e) {
                e.printStackTrace();
            }
            return true;
        });
    }

    public Spider chain(Function func) {
        //当前队列
        if (queues[index] == null) {
            queues[index] = new LinkedBlockingQueue(1000);
        }
        //下一个队列
        if (queues[index + 1] == null) {
            queues[index + 1] = new LinkedBlockingQueue(1000);
        }
        int k = index;
        tasks[index] = new Thread(
                () -> chain_process(func, k)
        );
        index++;
        return this;
    }

    private void chain_process(Function func, int k) {
        ExecutorService executorService = Executors.newCachedThreadPool();
        while (!queues[k].isEmpty() || isLastThreadLive(k)) {

            if (!queues[k].isEmpty()) {

                FetchParams fetchParams = null;
                try {
                    fetchParams = (FetchParams) func.apply(queues[k].take());
                } catch (InterruptedException e) {
                    System.out.println(e.getMessage());
                }
                FetchParams finalFetchParams = fetchParams;
                CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> fetchPageSource(finalFetchParams), executorService);
                future.thenAccept(pageSource -> {
                    //使用clazz
                    if (finalFetchParams.getClazz() != null) {
                        enqueueWithClazz(k, finalFetchParams, "" + pageSource);
                    }
                    //不使用clazz
                    if (finalFetchParams.getXpath() != null) {
                        List<String> newUrls = Xsoup.compile(finalFetchParams.getXpath()).evaluate(Jsoup.parse("" + pageSource)).list();
                        try {
                            for (int i = 0; i < newUrls.size(); i++) {
                                queues[k + 1].put(newUrls.get(i));
                            }
                        } catch (InterruptedException e) {
                            System.out.println(e.getMessage());

                        }
                    }
                });
            }
        }
//        executorService.shutdown();
//        try {
//            executorService.awaitTermination(Long.MAX_VALUE, TimeUnit.MINUTES);
//        } catch (InterruptedException e) {
//            e.printStackTrace();
//        }
    }

    private void enqueueWithClazz(int k, FetchParams fetchParams, String pageSource) {
        Class clazz = fetchParams.getClazz();
        ClassXpath classXpath = (ClassXpath) clazz.getAnnotation(ClassXpath.class);
        if (classXpath != null) {
            Elements elements = Xsoup.compile(classXpath.value()).evaluate(Jsoup.parse(pageSource)).getElements();
            elements.stream().forEach(
                    element -> {
                        fieldXpathProcess(queues[k + 1], element.outerHtml(), clazz);
                    });
        } else {
            fieldXpathProcess(queues[k + 1], pageSource, clazz);
        }
    }

    private void fieldXpathProcess(LinkedBlockingQueue queue, String pageSource, Class clazz) {
        Object obj = null;
        try {
            obj = clazz.newInstance();
        } catch (Exception e) {
            System.out.println(e.getMessage());
        }

        for (Field field : clazz.getDeclaredFields()) {
            Annotation[] annotations = field.getDeclaredAnnotations();
            if (annotations.length != 0) {
                Annotation annotation = annotations[0];
                if (annotation instanceof FieldXpath) {
                    Method setMethod = null;
                    try {
                        setMethod = clazz.getDeclaredMethod("set" + WordUtils.capitalize(field.getName()), String.class);//目前只支持String
                        List<String> lists = Xsoup.compile(((FieldXpath) annotation).value()).evaluate(Jsoup.parse(pageSource)).list();
                        setMethod.invoke(obj, "".join(",", lists));
                    } catch (Exception e) {
                        System.out.println(e.getMessage());
                    }
                }
            }
        }
        try {
            queue.put(obj);
        } catch (InterruptedException e) {
            System.out.println(e.getMessage());
        }
    }

    Random r = new Random();

    public String fetchPageSource(FetchParams fetchParams) {
//        double d2 = r.nextDouble() * 0.5 + 0.5;
//        try {
//            Thread.sleep((long) (d2 * 1000));
//        } catch (InterruptedException e) {
//            e.printStackTrace();
//        }
        String content = "";
        switch (fetchParams.getFlag()) {
            case simple: {
                try {
                    content = Util.req(fetchParams.getUrl());
                } catch (Exception e) {
                    //nothing to do
                }
            }
            break;
            case dynamic: {
                synchronized (this) {
                    try {
                        if (driver == null) {
                            driver = initWebDriver();
                        }
                        driver.get(fetchParams.getUrl());
                        driver.manage().timeouts().implicitlyWait(1, TimeUnit.SECONDS);
                        if (fetchParams.getScript() != null) {
                            try {
                                ((JavascriptExecutor) driver).executeScript(fetchParams.getScript());
                                if (fetchParams.getScriptLocator() != null) {
                                    new WebDriverWait(driver, fetchParams.getTimeout()).until(fetchParams.getScriptLocator());
                                }
                            } catch (Exception e) {
                                System.out.println(e.getMessage());
                            }
                        }
                        content = driver.getPageSource();
                    } catch (Exception e) {
                        System.out.println(e.getMessage());
                    }
                }
            }
            default:
                break;
        }
        return content;
    }

    Connection conn = null;

    public void initDB() {
        Yaml yaml = new Yaml();
        DBParams me = null;
        try {
            me = yaml.loadAs(new FileInputStream(new File("config/db.yaml")), DBParams.class);
        } catch (FileNotFoundException e) {
            e.printStackTrace();
        }
        System.out.println(me);

        try {
            //STEP 2: Register JDBC driver
            Class.forName("com.mysql.cj.jdbc.Driver");

            //STEP 3: Open a connection
            System.out.println("Connecting to a selected database...");
            conn = DriverManager.getConnection(me.getUrl(), me.getUsername(), me.getPassword());
            conn.setAutoCommit(true);
            System.out.println("Connected database successfully...");

            //STEP 4: Execute a query
            System.out.println("Inserting records into the table...");
            stmt = conn.createStatement();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
